package org.eventime

import org.FlinkStreamApp
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.bean.SensorReading
import org.diysource.SensorSource

/**
 * description ：diy 抽取时间戳和插入水位线
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 11:55
 * modified By ：
 * version:    : 1.0
 */
object DiyEventTimeExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    val stream = environment.addSource(new SensorSource)
      // 如果提前知道数据的时间戳是单调递增，也就是说没有乱序存在，可以用以下方法抽取时间戳
//       .assignAscendingTimestamps(r => r.ts)
      .assignTimestampsAndWatermarks(
        new PeriodicAssigner
      )
      .keyBy(_.id)
      .timeWindow(Time.seconds(5))
      .process(new WatermarkProcessFunction())
      .print()
  }

  class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading] {
    // 最大延迟时间是 1 秒钟
    val bound: Long = 5000
    // 用来保存观察到的最大事件时间戳
    var maxTs: Long = Long.MinValue

    override def getCurrentWatermark = new Watermark(maxTs - bound)

    // 每来到一个元素，都会调用一次, 抽取最大时间戳
    override def extractTimestamp(element: SensorReading, previousElementTimestamp: Long) = {
      maxTs = maxTs.max(element.ts)
      element.ts
    }
  }

  // [IN, OUT, KEY, W <: Window]
  class WatermarkProcessFunction extends ProcessWindowFunction[SensorReading, String, String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[String]) = {
      out.collect("窗口中共有： " + elements.size.toString + " 条数据")
    }
  }

}
